package com.jeffse.light.stream.core.client;

import com.jeffse.light.stream.core.constans.CommandConstans;
import com.jeffse.light.stream.core.data.stream.DataStream;
import com.jeffse.light.stream.core.handler.StreamClientHander;
import com.jeffse.light.stream.core.handler.decoder.DataStreamDecode;
import com.jeffse.light.stream.core.handler.encoder.DataStreamEncode;
import com.jeffse.light.stream.core.pojo.TranslateDataStream;
import com.jeffse.light.stream.core.thread.LightStreamThread;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;

@Slf4j
public class StreamClient implements Serializable {
    private ChannelFuture channelFuture = null;
    private static StreamClient INSTANCE = new StreamClient();
    private EventLoopGroup workerGroup = new NioEventLoopGroup();
    private String host = "localhost";
    private Integer port = 9123;

    /**
     * 启动客户端
     */
    public ChannelFuture start() {
        try {
            Bootstrap bootstrap = new Bootstrap();
            bootstrap.group(workerGroup)
                    .channel(NioSocketChannel.class)
                    .option(ChannelOption.SO_KEEPALIVE, true)
                    .handler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        public void initChannel(SocketChannel ch) {
                            ch.pipeline().addLast(new DataStreamEncode());
                            ch.pipeline().addLast(new DataStreamDecode());
                            ch.pipeline().addLast(new StreamClientHander());
                        }
                    });
            channelFuture = bootstrap.connect(host, port).sync();
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {

        }
        return channelFuture;
    }

    /**
     * 获取客户端
     * @return
     */
    public static StreamClient getClient(){
        return INSTANCE;
    }

    /**
     * 设置监听服务端 host
     * @param host
     * @return
     */
    public StreamClient setHost(String host){
        this.host = host;
        return this;
    }

    /**
     * 设置监听服务端 port
     * @param port
     * @return
     */
    public StreamClient setPort(Integer port){
        this.port = port;
        return this;
    }

    private ChannelFuture getChannelFuture(){
        if(channelFuture == null){
            this.start();
        }
        return channelFuture;
    }

    /**
     * 发送消息
     * @param dataStream
     */
    public void start(DataStream dataStream) throws InterruptedException {
        Channel channel = getChannelFuture().channel();
        channel.write(new TranslateDataStream(dataStream, CommandConstans.START, dataStream.getDataStreamName()));
        channel.flush();
    }

    /**
     * 发送消息
     * @param dataStreamName
     */
    public void cancel(String dataStreamName) throws InterruptedException {
        Channel channel = getChannelFuture().channel();
        channel.write(new TranslateDataStream(null, CommandConstans.STOP, dataStreamName));
        channel.flush();
    }

    /**
     * 发送消息
     * @param lightStreamThread
     */
    public void sendThread(LightStreamThread lightStreamThread) throws InterruptedException {
        getChannelFuture().channel().writeAndFlush(lightStreamThread);
    }


    /**
     * 停止客户端
     */
    public void stop() {
        workerGroup.shutdownGracefully();
    }
}
